package com.xu.t1.service;

import com.xu.t1.enums.DispatchResult;
import com.xu.t1.middleware.threadallocation.ThreadAllocationChainFactory;
import org.apache.commons.lang3.tuple.Pair;
import com.xu.t1.common.ListSplitter;
import com.xu.t1.common.RedisLockUtil;
import com.xu.t1.middleware.Thread.ThreadCounterService;
import com.xu.t1.model.LearningRecord;
import com.xu.t1.model.UploadLog;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.*;

/**
 * 将写入学习数据分配多线程异步写入
 *
 * @author xu
 *
 */


@Component
public class LearningRecordDispatcher {

    @Autowired
    @Qualifier("writeExecutor")
    private ExecutorService executor;

    @Autowired
    private LearningRecordService learningRecordService;

    @Autowired
    private ThreadCounterService threadCounterService;

    @Autowired
    private RedisLockUtil redisLockUtil;

    @Autowired
    private ThreadAllocationChainFactory threadAllocationChainFactory;


    public DispatchResult dispatch(List<LearningRecord> records, int totalSize, UploadLog log) {
        // 调用分布式锁+开线程异步写入
        List<Pair<Future<Boolean>, List<LearningRecord>>> futures = dispatchWithLock(records, totalSize, log);

        // 无线程资源或者没拿到锁
        if (futures.isEmpty()) return DispatchResult.BUSY;

        // 返回写入结果
        if (handleWriteAndCompensate(futures, records.get(0).getUserId(), totalSize, log)) {
            return DispatchResult.SUCCESS;
        } else return DispatchResult.FAILED;
    }

    // 分布式锁获取
    public List<Pair<Future<Boolean>, List<LearningRecord>>> dispatchWithLock(List<LearningRecord> records, int totalSize, UploadLog log) {
        return redisLockUtil.runWithLock(
                "dispatch_lock",        // 锁的key
                () -> dispatchTasks(records, totalSize, log),  // 要执行的任务
                Collections.emptyList()        // 如果没拿到锁，默认返回空列表
        );
    }


    // 开启线程 返回各线程异步结果引用
    public List<Pair<Future<Boolean>, List<LearningRecord>>> dispatchTasks(List<LearningRecord> records, int totalSize, UploadLog log) {
        // 线程动态分配算法
        int thread_count = threadAllocationChainFactory.createChain(totalSize);

        log.setThreadCount(thread_count);

        // 当算法计算无可用线程
        if (thread_count == 0) {
            return Collections.emptyList();
        }
        // 拆分链表
        List<List<LearningRecord>> partitions = ListSplitter.split(records, thread_count);

        List<Pair<Future<Boolean>, List<LearningRecord>>> futures = new ArrayList<>();

        for (List<LearningRecord> part : partitions) {
            try {
                // 向 Redis 发线程启动信号
                threadCounterService.updateThreadCount(1);
            } catch (Exception e) {
                continue;
            }

            // 提交异步任务
            Future<Boolean> future = executor.submit(() -> {
                boolean result = false;
                try {
                    result = learningRecordService.saveWithRetry(part);
                } finally {
                    // 向 Redis 发线程结束信号
                    threadCounterService.updateThreadCount(-1);
                }
                return result;
            });

            // 组合成 Pair 加入结果列表
            futures.add(Pair.of(future, part));
        }

        return futures;
    }

    // 处理异步处理结果
    public boolean handleWriteAndCompensate(List<Pair<Future<Boolean>, List<LearningRecord>>> taskResults, Long userId, int totalSize, UploadLog log) {
        boolean hasFailure = false;

        for (Pair<Future<Boolean>, List<LearningRecord>> pair : taskResults) {
            try {
                Future<Boolean> future = pair.getLeft();
                Boolean result = future.get(); // 阻塞
                if (!Boolean.TRUE.equals(result)) {
                    hasFailure = true;
                    break;
                }
            } catch (Exception e) {
                hasFailure = true;
                break;
            }
        }

        if (hasFailure) {

            // 写入失败启动补偿删除
            deleteRecords(taskResults);
            return false;
        } else {
            // 全部写入成功 更新用户学习记录数量
            try {
                learningRecordService.updateRecordCountWithLockAndRetry(userId, totalSize, log);
                return true;
            } catch (Exception e) {

                // 更新失败 同样进行删除
                deleteRecords(taskResults);
                return false;
            }
        }
    }

    // 删除补偿
    public void deleteRecords(List<Pair<Future<Boolean>, List<LearningRecord>>> taskResults) {
        for (Pair<Future<Boolean>, List<LearningRecord>> pair : taskResults) {
            for (LearningRecord record : pair.getRight()) {
                if (record.getId() != null) {
                    learningRecordService.deleteRecords(record.getId());
                }
            }
        }
    }

}